package org.jetlinks.pro.cassandra.metadata;

import lombok.AllArgsConstructor;
import org.hswebframework.ezorm.core.ValueCodec;
import org.hswebframework.ezorm.core.meta.ObjectMetadata;
import org.hswebframework.ezorm.rdb.executor.SqlRequests;
import org.hswebframework.ezorm.rdb.executor.reactive.ReactiveSqlExecutor;
import org.hswebframework.ezorm.rdb.executor.wrapper.ColumnWrapperContext;
import org.hswebframework.ezorm.rdb.executor.wrapper.ResultWrapper;
import org.hswebframework.ezorm.rdb.executor.wrapper.ResultWrappers;
import org.hswebframework.ezorm.rdb.metadata.*;
import org.hswebframework.ezorm.rdb.metadata.parser.TableMetadataParser;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;
import java.util.Optional;

/**
 * Cassandra表结构解析器,用于获取已存在的cassandra表信息.
 * <p>
 * Cassandra的表结构信息存储在: system_schema.tables 和 system_schema.columns 中.
 *
 * @author zhouhao
 * @see TableMetadataParser
 * @see RDBSchemaMetadata#getTableReactive(String)
 * @since 1.10
 */
@AllArgsConstructor
public class CassandraTableMetadataParser implements TableMetadataParser {

    //schema
    private final RDBSchemaMetadata schema;

    /**
     * @return ReactiveSqlExecutor
     * @see CassandraReactiveSqlExecutor
     */
    public ReactiveSqlExecutor getSqlExecutor() {
        return schema.findFeatureNow(ReactiveSqlExecutor.ID);
    }

    @Override
    public Flux<String> parseAllTableNameReactive() {
        return this
            .getSqlExecutor()
            //查询system_schema.tables表
            .select("select table_name as name from system_schema.tables where keyspace_name=?", schema.getName())
            .map(map -> String.valueOf(map.get("name")));
    }


    @Override
    public Mono<Boolean> tableExistsReactive(String name) {
        return this
            .getSqlExecutor()
            .select("select table_name as name from system_schema.tables where keyspace_name=? and table_name = ?", schema.getName(), name)
            .hasElements();
    }


    @Override
    public Mono<? extends ObjectMetadata> parseByNameReactive(String name) {
        RDBTableMetadata tableMetadata = schema.newTable(name);
        tableMetadata.setName(name);
        return this
            .getSqlExecutor()
            //查询表结构
            .select(SqlRequests
                        .of(
                            "select column_name as name,type as type from system_schema.columns where table_name = ? and keyspace_name = ?",
                            name, schema.getName()
                        ),
                    new CassandraColumnResultWrapper(tableMetadata))
            .doOnNext(tableMetadata::addColumn)
            .then(Mono.just(tableMetadata))
            //没有列则认为表不存在
            .filter(table -> table.getColumns().size() > 0);
    }

    @Override
    public Flux<? extends ObjectMetadata> parseAllReactive() {
        return parseAllTableNameReactive()
            .flatMap(this::parseByNameReactive);
    }

    @AllArgsConstructor
    static class CassandraColumnResultWrapper implements ResultWrapper<RDBColumnMetadata, Void> {
        private final RDBTableMetadata table;

        @Override
        public RDBColumnMetadata newRowInstance() {
            return table.newColumn();
        }

        @Override
        public void wrapColumn(ColumnWrapperContext<RDBColumnMetadata> context) {
            String column = context.getColumnLabel();
            String value = String.valueOf(context.getResult());

            if ("name".equalsIgnoreCase(column)) {
                context.getRowInstance().setName(value);
            } else if ("type".equalsIgnoreCase(column)) {
                context.getRowInstance().setType(table.getDialect().convertDataType(value));
            }
        }

        @Override
        public boolean completedWrapRow(RDBColumnMetadata column) {
            //设置ValueCodec,用于在crud操作时进行数据转换

            if (column.getType() instanceof ValueCodec) {
                column.setValueCodec(((ValueCodec<?, ?>) column.getType()));
            } else {
                column.findFeature(ValueCodecFactory.ID)
                      .flatMap(factory -> factory.createValueCodec(column))
                      .ifPresent(column::setValueCodec);
            }
            return true;
        }

        @Override
        public Void getResult() {
            return null;
        }
    }

    @Override
    public Optional<? extends ObjectMetadata> parseByName(String name) {
        //暂不支持非响应式
        throw new UnsupportedOperationException();
    }

    @Override
    public List<? extends ObjectMetadata> parseAll() {
        //暂不支持非响应式
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean tableExists(String name) {
        //暂不支持非响应式
        throw new UnsupportedOperationException();
    }

    @Override
    public List<String> parseAllTableName() {
        //暂不支持非响应式
        throw new UnsupportedOperationException();
    }

}
